// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#include "coin2/exchange/base/executor/live_executor.h"

#include <memory>
#include <vector>

namespace coin2::exchange::base::executor {

void LiveAsyncExecutor::Start() {
  StartAsync();
  topic_distributor_.LogTopic();

  std::thread stats_thread = std::thread([&]() { stats_ioc_.run(); });
  std::vector<std::thread> t;
  // reserve 1 thread for dequeue
  for (size_t i = 0; i < config_.raw_threads(); i++) {
    t.emplace_back([&]() { raw_ioc_.run(); });
  }

  std::vector<std::thread> pasrer_threads;
  std::unique_ptr<boost::asio::io_context::work> parser_work;
  if (parser_ioc_ != nullptr) {
    parser_work = std::make_unique<boost::asio::io_context::work>(*parser_ioc_);
  }
  for (size_t i = 0; i < config_.parser_threads(); i++) {
    pasrer_threads.emplace_back([&]() { parser_ioc_->run(); });
  }

  // Compress and archive process need to be distributed due to the traffic
  // load is approaching to the limit of single thread capacity.
  std::unique_ptr<boost::asio::io_context::work> writer_work;
  if (writer_ioc_ != nullptr) {
    writer_work = std::make_unique<boost::asio::io_context::work>(*writer_ioc_);
  }
  std::vector<std::thread> writer_threads;
  for (size_t i = 0; i < config_.writer_threads(); ++i) {
    writer_threads.emplace_back([&]() { writer_ioc_->run(); });
  }

  boost::asio::io_context::work work{main_ioc_};
#ifdef DISABLE_REACTOR_INTEGRATION
  main_ioc_.run();
#else
  CHECK_EQ(main_reactor_.Start(), absl::OkStatus());
  CHECK_EQ(main_reactor_.WaitForStopped(), absl::OkStatus());
#endif

  StopAsync();

  stats_thread.join();
  for (size_t i = 0; i < config_.raw_threads(); i++) {
    t[i].join();
  }
  for (size_t i = 0; i < config_.writer_threads(); ++i) {
    writer_threads[i].join();
  }
}

void LiveAsyncExecutor::Stop() {
#ifndef DISABLE_REACTOR_INTEGRATION
  CHECK_EQ(main_reactor_.Stop(), absl::OkStatus());
#endif

  main_ioc_.stop();
  raw_ioc_.stop();
  stats_ioc_.stop();
  if (writer_ioc_ != nullptr) {
    writer_ioc_->stop();
  }
}

void LiveAsyncExecutor::OnTopicRecord(
    std::shared_ptr<TopicRecordPacket> packet) {
  topic_distributor_.OnTopicRecord(packet->topic_name, packet.get());
  if (counter_) {
    counter_->ProcessOneMsg();
  }
}

}  // namespace coin2::exchange::base::executor
